package com.youshang520i.zookeeper;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * @author youshang
 *  长久数据监控
 */
public class Zookeeper{

    private static final Logger logger = LoggerFactory.getLogger(Zookeeper.class);
    private final static String CONNECTSTRING = "192.168.3.123:2181";
    private final static int SESSION_TIMEOUT = 50 * 1000;
    private final static String PATH = "/youshang";
    private ZooKeeper zk = null;
    private String oldValue = null;

    public ZooKeeper startZK() throws Exception {
        return new ZooKeeper(CONNECTSTRING, SESSION_TIMEOUT, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("回调接口");
            }
        });
    }

    public void stopZK() throws Exception {
        if(zk != null) {
            zk.close();
        }
    }

    /**
     * 创建znode节点
     * @param nodePath 创建的路径
     * @param nodeValue
     * @throws Exception
     */
    public void createZnode(String nodePath, String nodeValue) throws Exception {
        // OPEN_ACL_UNSAFE:类似关闭防火墙
        // CreateMode.PERSISTENT: 创建的节点由Java控制是持久的
        zk.create(nodePath, nodeValue.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    public String getZnode(String nodePath) throws Exception {
        String result = null;
        byte[] byteArray = zk.getData(PATH, new Watcher() { // 一次性触发
            @Override
            public void process(WatchedEvent event) {
                try {
                    trigerValue(PATH);
                }  catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, new Stat());
        result = new String(byteArray); // 转为字符串
        oldValue = result; // 一开始初始值
        return result;
    }

    // 每次所观察的节点改变就触发
    public boolean trigerValue(final String nodePath) throws Exception {
        String result = null;
        byte[] byteArray = zk.getData(PATH, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                try {
                    trigerValue(nodePath);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, new Stat());
        result = new String(byteArray); // 转为字符串
        String newValue = result;
        if(oldValue.equals(newValue)){
            logger.info("*********no changes");
            return false;
        }else{
            logger.info("************oldValue:"+oldValue+"\t newValue:"+newValue);
            oldValue = newValue;
            return true;
        }
    }
    // 监控节点 /atguigu下的数据
    public static void main(String[] args) throws Exception {
        Zookeeper watchOne = new Zookeeper();
        watchOne.setZk(watchOne.startZK());
        if(watchOne.getZk().exists(PATH, false) == null){ // 没有该节点才能创建，不然会报错，节点是不能覆盖的
            watchOne.createZnode( PATH, "AAA");
            String retValue = watchOne.getZnode(PATH);
            logger.info("retValue="+retValue);
            Thread.sleep(Long.MAX_VALUE);
        }else{
            logger.info("I have no znode");
        }
    }
    public ZooKeeper getZk() {
        return zk;
    }
    public void setZk(ZooKeeper zk) {
        this.zk = zk;
    }

    public String getOldValue() {
        return oldValue;
    }

    public void setOldValue(String oldValue) {
        this.oldValue = oldValue;
    }

}
